package com.gin.demo;

import lombok.*;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.io.IOException;

/**
 * 场景:
 * 先以IP维度计算IP关联次数, 再以userId维度计算userId关联次数
 * 两个结果分别进行sink
 *
 * @author gin
 * @date 2021/5/31
 * <p>
 * <p>
 * 测试数据
10.30.10.2,80081
10.30.10.2,80082
10.30.10.2,80083
10.30.10.2,80083
10.30.10.2,80083
 */
public class MultipleSinkDemo {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        final OutputTag<BrowseVO> ipSideOutput = new
                OutputTag<>("ipSideOutput", TypeInformation.of(BrowseVO.class));
        // 连接socket获取输入的数据
        // nc -lk 8888
        // netstat -natp | grep 8888
        DataStreamSource<String> socketStream = env.socketTextStream("172.21.194.111", 8888, "\n");
        SingleOutputStreamOperator<BrowseVO> ipCountStream = socketStream
                .map(new MapFunction<String, BrowseVO>() {
                    @Override
                    public BrowseVO map(String line) {
                        String[] split = line.split(",");
                        return new BrowseVO(split[0], split[1]);
                    }
                })
                .keyBy(BrowseVO::getIp)
                .process(new KeyedProcessFunction<String, BrowseVO, BrowseVO>() {

                    ValueState<BrowseVO> ipCountState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //1天过期(1天内不会重复消费)
                        StateTtlConfig stateTtlConfig = StateTtlConfig.newBuilder(Time.days(1))
                                //过期不可访问
                                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                                //在状态值被创建和被更新时重设TTL
                                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                                .build();

                        ValueStateDescriptor<BrowseVO> ipCountStateDesc = new ValueStateDescriptor<>(
                                "ipCountState",
                                BrowseVO.class
                        );
                        ipCountStateDesc.enableTimeToLive(stateTtlConfig);
                        ipCountState = getRuntimeContext().getState(ipCountStateDesc);
                    }

                    @Override
                    public void processElement(BrowseVO value, Context ctx, Collector<BrowseVO> out) {
                        try {
                            BrowseVO ipVo = ipCountState.value();
                            //不存在的key才继续收集
                            if (ipVo == null) {
                                ipVo = BrowseVO.builder()
                                        .ip(value.ip)
                                        .count(1L)
                                        .build();
                            } else {
                                ipVo.setCount(ipVo.count + 1L);
                            }
                            //更新为存在
                            ipCountState.update(ipVo);
                            ctx.output(ipSideOutput, ipVo);
                            out.collect(value);
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                });

        //侧输出流需要通过tag获取, 存储IP维度数据
        ipCountStream
                .getSideOutput(ipSideOutput)
                .addSink(new PrintSinkFunction<>());

        //默认打印主输出流(用户统计维度)
        ipCountStream
                .keyBy(BrowseVO::getUserId)
                .process(new KeyedProcessFunction<String, BrowseVO, BrowseVO>() {

                    ValueState<BrowseVO> userCountState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //1天过期(1天内不会重复消费)
                        StateTtlConfig stateTtlConfig = StateTtlConfig.newBuilder(Time.days(1))
                                //过期不可访问
                                .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
                                //在状态值被创建和被更新时重设TTL
                                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                                .build();

                        ValueStateDescriptor<BrowseVO> ipCountStateDesc = new ValueStateDescriptor<>(
                                "ipCountState",
                                BrowseVO.class
                        );
                        ipCountStateDesc.enableTimeToLive(stateTtlConfig);
                        userCountState = getRuntimeContext().getState(ipCountStateDesc);
                    }

                    @Override
                    public void processElement(BrowseVO value, Context ctx, Collector<BrowseVO> out) throws Exception {
                        try {
                            BrowseVO userVo = userCountState.value();
                            //不存在的key才继续收集
                            if (userVo == null) {
                                userVo = BrowseVO.builder()
                                        .userId(value.userId)
                                        .count(1L)
                                        .build();
                            } else {
                                userVo.setCount(userVo.count + 1L);
                            }
                            //更新为存在
                            userCountState.update(userVo);
                            out.collect(userVo);
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                })
                .addSink(new PrintSinkFunction<>());

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }


}

@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
@ToString
class BrowseVO {

    public BrowseVO(String ip, String userId) {
        this.ip = ip;
        this.userId = userId;
    }

    public String ip;

    public String userId;

    public long count;

}
